package org.apache.rocketmq.proxy.service.route;



import com.google.common.base.MoreObjects;
import com.google.common.base.Preconditions;
import com.google.common.math.IntMath;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.client.latency.MQFaultStrategy;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.route.QueueData;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/17
 * @方法描述：消息队列选择器，该类的对象负责根据主题路由信息构建消息队列列表，并提供队列选择策略
 * 在生产者客户端发送消息的时候，消息队列选择器就会选择合适的目标消息队列
 */
public class MessageQueueSelector {

    //Broker代理队列的Id，暂时还不必关注这个代理队列
    private static final int BROKER_ACTING_QUEUE_ID = -1;

    //存储正常消息队列的集合
    private final List<AddressableMessageQueue> queues = new ArrayList<>();

    //存储Broker代理队列的集合,代理队列这个概念等到第十九版本代码我再为大家解释代理队列的概念和作用
    //现在就算解释，大家可能也不会很清楚，等待第十九版本代码用到了这个代理队列，结合具体例子一解释大家就全明白了
    //现在先混个眼熟即可，或者大家先记住这样一个设定：只要客户端指定了消息要发送或者消费的Broker节点的名称，那就根据broker节点的名称从brokerNameQueueMap中获取对应消息队列即可
    private final List<AddressableMessageQueue> brokerActingQueues = new ArrayList<>();

    //key为Broker节点的名称，value为对应的消息队列对象
    private final Map<String, AddressableMessageQueue> brokerNameQueueMap = new ConcurrentHashMap<>();

    //普通队列索引计数器
    private final AtomicInteger queueIndex;

    //Broker代理队列索引计数器
    private final AtomicInteger brokerIndex;

    //消息队列故障转移策略组件，在第十二版本代码中还用不到，暂时不必关注
    private MQFaultStrategy mqFaultStrategy;

    //构造方法
    public MessageQueueSelector(TopicRouteWrapper topicRouteWrapper, MQFaultStrategy mqFaultStrategy, boolean read) {
        //判断要构建的消息对立的权限，然后决定创建读队列还是写队列
        if (read) {
            //如果是读权限，那么就创建读队列
            this.queues.addAll(buildRead(topicRouteWrapper));
        } else {//否则创建写队列
            this.queues.addAll(buildWrite(topicRouteWrapper));
        }
        //构建Broker代理队列
        buildBrokerActingQueues(topicRouteWrapper.getTopicName(), this.queues);
        //创建随机数生成器
        Random random = new Random();
        //创建队列索引计数器，初始化队列索引为随机值，这样可以避免所有客户端都选择同一个队列处理消息
        this.queueIndex = new AtomicInteger(random.nextInt());
        //初始化Broker索引为随机值，这样可以避免所有客户端都选择同一个Broker处理消息
        this.brokerIndex = new AtomicInteger(random.nextInt());
        this.mqFaultStrategy = mqFaultStrategy;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：构建读权限消息队列的方法
     */
    private static List<AddressableMessageQueue> buildRead(TopicRouteWrapper topicRoute) {

        //创建一个AddressableMessageQueue集合，AddressableMessageQueue中存储的就是具体的队列和队列所属的Broker节点的网络地址
        Set<AddressableMessageQueue> queueSet = new HashSet<>();
        //从路由信息中获取队列数据集合
        List<QueueData> qds = topicRoute.getQueueDatas();
        //如果队列数据集合为空，返回一个空的队列列表
        if (qds == null) {
            return new ArrayList<>();
        }
        //遍历队列数据集合
        for (QueueData qd : qds) {
            //检查当前遍历到的队列是否具备读权限
            if (PermName.isReadable(qd.getPerm())) {
                //如果具备读权限，那就获得队列所属Broker节点的名称，然后根据Broker节点名称得到对应的网络地址
                //注意，这里得到的是Broker节点所在主从组的主节点的网络地址
                String brokerAddr = topicRoute.getMasterAddrPrefer(qd.getBrokerName());
                //如果Broker节点的网络地址为空，则跳过当前队列
                if (brokerAddr == null) {
                    continue;
                }
                //接下来就可以为当前遍历到的队列创建一个可寻址的消息队列对象，也就是AddressableMessageQueue对象
                for (int i = 0; i < qd.getReadQueueNums(); i++) {
                    //创建AddressableMessageQueue对象，把主题名、Broker名称和队列Id等信息封装成MessageQueue对象
                    //然后把MessageQueue对象和Broker主节点网络地址设置到AddressableMessageQueue对象中
                    AddressableMessageQueue mq = new AddressableMessageQueue(new MessageQueue(topicRoute.getTopicName(), qd.getBrokerName(), i), brokerAddr);
                    //添加到队列集合中
                    queueSet.add(mq);
                }
            }
        }
        //消息队列排序，然后返回，因为没有定义排序比较器，所以就会使用AddressableMessageQueue对象自己的compareTo方法进行排序
        return queueSet.stream().sorted().collect(Collectors.toList());
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：构建写权限消息队列的方法
     */
    private static List<AddressableMessageQueue> buildWrite(TopicRouteWrapper topicRoute) {

        Set<AddressableMessageQueue> queueSet = new HashSet<>();
        //判断当前主题是否为顺序发送消息主题
        if (StringUtils.isNotBlank(topicRoute.getOrderTopicConf())) {
            //如果是顺序主题，那就得到顺序主题的配置信息，顺序主题的配置信息是一个字符串，格式为："brokerName1:队列数量;brokerName2:队列数量
            String[] brokers = topicRoute.getOrderTopicConf().split(";");
            //遍历顺序主题的配置信息
            for (String broker : brokers) {
                //得到broker名称和队列数量字符串
                String[] item = broker.split(":");
                //得到Broker名称
                String brokerName = item[0];
                //得到该Broker在主从组中的主节点地址
                String brokerAddr = topicRoute.getMasterAddr(brokerName);
                //如果没有找到Broker地址，跳过当前Broker
                if (brokerAddr == null) {
                    continue;
                }
                //得到当前Broker的队列数量
                int nums = Integer.parseInt(item[1]);
                //为当前Broker创建指定数量的消息队列
                for (int i = 0; i < nums; i++) {
                    AddressableMessageQueue mq = new AddressableMessageQueue(new MessageQueue(topicRoute.getTopicName(), brokerName, i), brokerAddr);
                    queueSet.add(mq);
                }
            }
        } else {
            //下面代码的逻辑和buildRead()方法几乎一致，所以我就不添加注释了，大家简单看看就行
            List<QueueData> qds = topicRoute.getQueueDatas();
            if (qds == null) {
                return new ArrayList<>();
            }
            for (QueueData qd : qds) {
                if (PermName.isWriteable(qd.getPerm())) {
                    String brokerAddr = topicRoute.getMasterAddr(qd.getBrokerName());
                    if (brokerAddr == null) {
                        continue;
                    }
                    for (int i = 0; i < qd.getWriteQueueNums(); i++) {
                        AddressableMessageQueue mq = new AddressableMessageQueue(new MessageQueue(topicRoute.getTopicName(), qd.getBrokerName(), i), brokerAddr);
                        queueSet.add(mq);
                    }
                }
            }
        }
        return queueSet.stream().sorted().collect(Collectors.toList());
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：构建Broker代理队列的方法，该方法暂时不必关注
     */
    private void buildBrokerActingQueues(String topic, List<AddressableMessageQueue> normalQueues) {
        for (AddressableMessageQueue mq : normalQueues) {
            AddressableMessageQueue brokerActingQueue = new AddressableMessageQueue(new MessageQueue(topic, mq.getMessageQueue().getBrokerName(), BROKER_ACTING_QUEUE_ID), mq.getBrokerAddr());
            if (!brokerActingQueues.contains(brokerActingQueue)) {
                brokerActingQueues.add(brokerActingQueue);
                brokerNameQueueMap.put(brokerActingQueue.getBrokerName(), brokerActingQueue);
            }
        }
        Collections.sort(brokerActingQueues);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据brokerName获取对应的代理队列的方法
     */
    public AddressableMessageQueue getQueueByBrokerName(String brokerName) {
        return this.brokerNameQueueMap.get(brokerName);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：选择一个消息队列的方法，方法参数表示是否使用代理队列
     */
    public AddressableMessageQueue selectOne(boolean onlyBroker) {
        //根据方法参数确定选择消息队列的索引，如果使用代理队列，那就使用broker索引，否则使用queue索引
        int nextIndex = onlyBroker ? brokerIndex.getAndIncrement() : queueIndex.getAndIncrement();
        //根据索引选择具体队列的方法
        return selectOneByIndex(nextIndex, onlyBroker);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：选择具体的消息策略的方法
     */
    public AddressableMessageQueue selectOneByPipeline(boolean onlyBroker) {
        //先判断程序内部是否启动了故障转移策略，以及故障转移组件是否不为空
        //在源码中并没有启动这个故障转移策略，之前我也跟大家解释了，只有在集群模式部署proxy的时候，才会启动这个策略
        //所以我们的代码并不会进入下面的分支，但既然代码都已经引入了，稍微加点注释也不是不行
        if (mqFaultStrategy != null && mqFaultStrategy.isSendLatencyFaultEnable()) {
            //存储待选则的消息队列的集合
            List<MessageQueue> messageQueueList = null;
            //存储最终选择的消息队列
            MessageQueue messageQueue = null;
            //判断是使用代理队列还是普通队列
            if (onlyBroker) {
                //如果使用代理队列，就把代理队列列表转换成普通队列列表，之所以先进行转换
                //是应为接下来故障转移组件需要使用普通队列列表进行队列筛选
                messageQueueList = transferAddressableQueues(brokerActingQueues);
            } else {
                //如果不使用代理队列，就把普通队列列表转换成代理队列列表
                messageQueueList = transferAddressableQueues(queues);
            }
            //存储最终选择的可寻址的消息队列，这个队列就是最后返回给用户的队列
            AddressableMessageQueue addressableMessageQueue = null;
            //在这里使用故障转移策略中的过滤器选择可用的消息队列，注意，这里使用了两个过滤器，一个是Broker节点的可用性过滤器，一个是队列的可达性过滤器
            messageQueue = selectOneMessageQueue(messageQueueList, onlyBroker ? brokerIndex : queueIndex,
                    mqFaultStrategy.getAvailableFilter(), mqFaultStrategy.getReachableFilter());
            //把选中的队列转换成AddressableMessageQueue对象
            addressableMessageQueue = transferQueue2Addressable(messageQueue);
            //如果找到了目标队列，那就直接返回即可
            if (addressableMessageQueue != null) {
                return addressableMessageQueue;
            }
            //如果没有找到目标队列，那就放宽限制，只使用Broker可用过滤器筛选队列
            //这可能会选择到网络连接不太稳定但仍在运行的Broker节点
            messageQueue = selectOneMessageQueue(messageQueueList, onlyBroker ? brokerIndex : queueIndex, mqFaultStrategy.getAvailableFilter());
            //把选中的队列转换成AddressableMessageQueue对象
            addressableMessageQueue = transferQueue2Addressable(messageQueue);
            //如果找到了目标队列，那就直接返回即可
            if (addressableMessageQueue != null) {
                return addressableMessageQueue;
            }
            //如果没有找到目标队列，那就放宽限制，只使用Broker可达过滤器筛选队列
            messageQueue = selectOneMessageQueue(messageQueueList, onlyBroker ? brokerIndex : queueIndex, mqFaultStrategy.getReachableFilter());
            //把选中的队列转换成AddressableMessageQueue对象
            addressableMessageQueue = transferQueue2Addressable(messageQueue);
            //如果找到了目标队列，那就直接返回即可
            if (addressableMessageQueue != null) {
                return addressableMessageQueue;
            }
        }
        //执行到这里就意味着上面的分支并没有选择到可用的消息队列，也可能意味着程序并没有启动故障转移机制
        //总之这里就是一个托底的保障机制，在没有选择到队列的情况下，直接使用普通方式选择一个消息队列返回即可
        return selectOne(onlyBroker);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：使用过滤器选择消息队列的方法
     */
    private MessageQueue selectOneMessageQueue(List<MessageQueue> messageQueueList, AtomicInteger sendQueue, TopicPublishInfo.QueueFilter...filter) {

        //对消息队列集合判空
        if (messageQueueList == null || messageQueueList.isEmpty()) {
            return null;
        }
        //对过滤器判空
        if (filter != null && filter.length != 0) {
            for (int i = 0; i < messageQueueList.size(); i++) {
                //计算要获得的下一个队列的索引
                int index = Math.abs(sendQueue.incrementAndGet() % messageQueueList.size());
                //得到消息队列
                MessageQueue mq = messageQueueList.get(index);
                boolean filterResult = true;
                //遍历所有过滤器
                for (TopicPublishInfo.QueueFilter f: filter) {
                    Preconditions.checkNotNull(f);
                    //使用过滤器筛选消息队列
                    filterResult &= f.filter(mq);
                }
                //如果所有过滤器都通过，则返回该消息队列
                if (filterResult) {
                    return mq;
                }
            }
        }
        //如果遍历完所有队列都没有找到符合条件的，或者没有提供过滤器
        //则返回null表示没有符合条件的队列
        return null;
    }


    public List<MessageQueue> transferAddressableQueues(List<AddressableMessageQueue> addressableMessageQueueList) {
        if (addressableMessageQueueList == null) {
            return null;
        }
        return addressableMessageQueueList.stream()
                .map(AddressableMessageQueue::getMessageQueue)
                .collect(Collectors.toList());
    }


    private AddressableMessageQueue transferQueue2Addressable(MessageQueue messageQueue) {
        for (AddressableMessageQueue amq: queues) {
            if (amq.getMessageQueue().equals(messageQueue)) {
                return amq;
            }
        }
        return null;
    }


    public AddressableMessageQueue selectNextOne(AddressableMessageQueue last) {
        boolean onlyBroker = last.getQueueId() < 0;
        AddressableMessageQueue newOne = last;
        int count = onlyBroker ? brokerActingQueues.size() : queues.size();
        for (int i = 0; i < count; i++) {
            newOne = selectOne(onlyBroker);
            if (!newOne.getBrokerName().equals(last.getBrokerName()) || newOne.getQueueId() != last.getQueueId()) {
                break;
            }
        }
        return newOne;
    }


    //以下方法的逻辑都很简单，我就不添加详细注释了，大家简单看看就行

    public AddressableMessageQueue selectOneByIndex(int index, boolean onlyBroker) {
        if (onlyBroker) {
            if (brokerActingQueues.isEmpty()) {
                return null;
            }
            return brokerActingQueues.get(IntMath.mod(index, brokerActingQueues.size()));
        }

        if (queues.isEmpty()) {
            return null;
        }
        return queues.get(IntMath.mod(index, queues.size()));
    }


    public List<AddressableMessageQueue> getQueues() {
        return queues;
    }


    public List<AddressableMessageQueue> getBrokerActingQueues() {
        return brokerActingQueues;
    }


    public MQFaultStrategy getMQFaultStrategy() {
        return mqFaultStrategy;
    }


    public void setMQFaultStrategy(MQFaultStrategy mqFaultStrategy) {
        this.mqFaultStrategy = mqFaultStrategy;
    }


    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (!(o instanceof MessageQueueSelector)) {
            return false;
        }
        MessageQueueSelector queue = (MessageQueueSelector) o;
        return Objects.equals(queues, queue.queues) &&
                Objects.equals(brokerActingQueues, queue.brokerActingQueues);
    }


    @Override
    public int hashCode() {
        return Objects.hash(queues, brokerActingQueues);
    }


    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("queues", queues)
                .add("brokerActingQueues", brokerActingQueues)
                .add("brokerNameQueueMap", brokerNameQueueMap)
                .add("queueIndex", queueIndex)
                .add("brokerIndex", brokerIndex)
                .toString();
    }
}
